package org.example.idempotent.mq;

import lombok.RequiredArgsConstructor;
import org.aspectj.lang.ProceedingJoinPoint;
import org.aspectj.lang.annotation.Around;
import org.aspectj.lang.annotation.Aspect;
import org.aspectj.lang.reflect.MethodSignature;
import org.example.idempotent.ExpressionUtils;
import org.example.idempotent.RedisScriptCache;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.script.RedisScript;
import org.springframework.stereotype.Component;
import java.lang.reflect.Method;
import java.util.Collections;
import java.util.Objects;
import java.util.concurrent.TimeUnit;

/**
 * 防止消息队列消费者重复消费消息切面控制器
 */
@Aspect
@RequiredArgsConstructor
@Component
public class NoMQDuplicateConsumeAspect {
    private static final Logger log = LoggerFactory.getLogger(NoMQDuplicateConsumeAspect.class);

    private final RedisTemplate<String, Object> redisTemplate;

    // 直接使用预加载的Lua脚本（启动时已初始化）
    private static final RedisScript<String> IDEMPOTENT_MQ_SCRIPT = RedisScriptCache.getIdempotentMqScript();


    @Around("@annotation(noMQDuplicateConsume)")
    public Object noRepetitionConsume(ProceedingJoinPoint joinPoint, NoMQDuplicateConsume noMQDuplicateConsume){
        final String lockKey = buildLockKey(joinPoint, noMQDuplicateConsume);
        final long keyTimeout = noMQDuplicateConsume.keyTimeout();
        long keyTimeoutMillis = TimeUnit.SECONDS.toMillis(keyTimeout);
        //执行Lua脚本：原子性判断+设置锁
        String oldValue = redisTemplate.execute(
                IDEMPOTENT_MQ_SCRIPT,
                Collections.singletonList(lockKey),
                IdempotentMQConsumeStatusEnum.CONSUMING.getCode(),
                keyTimeoutMillis
        );

        // 判断是否重复消费
        if (Objects.nonNull(oldValue)) {
            boolean isConsuming = IdempotentMQConsumeStatusEnum.isConsuming(oldValue);
            log.warn("[{}] MQ重复消费，当前状态：{}", lockKey,
                    isConsuming ? "消费中（可能并发）" : "已消费完成");

            // 若状态为"消费中"，抛出异常触发MQ重试（避免消息丢失）
            if (isConsuming) {
                throw new RuntimeException(String.format("消息幂等校验失败：%s 正在消费中", lockKey));
            }
            // 若已消费完成，直接返回（不重复处理）
            return null;
        }

        Object result;
        try {
            // 执行标记了消息队列防重复消费注解的方法原逻辑
            result = joinPoint.proceed();
            // 业务成功：更新状态为"已消费"，保持相同过期时间
            redisTemplate.opsForValue().set(lockKey, IdempotentMQConsumeStatusEnum.CONSUMED.getCode(), keyTimeout, TimeUnit.SECONDS);
        } catch (Throwable ex) {
            // 业务失败：删除锁，允许MQ重试（确保失败消息可重新处理）
            boolean deleteSuccess = redisTemplate.delete(lockKey);
            log.error("[{}] 消息消费失败，删除幂等锁：{}", lockKey, deleteSuccess, ex);
            // 包装异常，保留原始堆栈（方便排查）
            throw new RuntimeException("业务处理失败，已释放幂等锁允许重试", ex);
        }
        return result;



    }


    private String buildLockKey(ProceedingJoinPoint joinPoint, NoMQDuplicateConsume noMQDuplicateConsume){

        StringBuilder keyBuilder =
                new StringBuilder(noMQDuplicateConsume.prefix())
                        .append(NoMqDuplicateConsumeConstant.REDIS_SEPARATOR)
                        .append(getArgsValueForKey(joinPoint, noMQDuplicateConsume.uniqueKey()));
        return keyBuilder.toString();

    }


    private String getArgsValueForKey(ProceedingJoinPoint joinPoint, String uniqueKey){
        Object[] args = joinPoint.getArgs();
        MethodSignature methodSignature = (MethodSignature) joinPoint.getSignature();
        Method method = methodSignature.getMethod();
        String argValue = ExpressionUtils.getExpressionValueAliasAble(args, method, uniqueKey);
        return argValue;
    }





}
